package com.lhd.app.metrics;

import com.lhd.common.utils.DateFormatUtil;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 全窗口函数
 */
public class GoodsMetricWindowFunction extends ProcessAllWindowFunction<GoodsMetricsAccumulator, GoodsMetricsResult, TimeWindow> {

    @Override
    public void process(Context context, Iterable<GoodsMetricsAccumulator> elements,
                        Collector<GoodsMetricsResult> out) throws Exception {

        GoodsMetricsAccumulator accumulator = elements.iterator().next();
        TimeWindow window = context.window();

        GoodsMetricsResult result = new GoodsMetricsResult(
                DateFormatUtil.toYmdHms(window.getStart()),
                DateFormatUtil.toYmdHms(window.getEnd()),
                accumulator.mids.size(),      // UV
                accumulator.pv,               // PV
                accumulator.cartMids.size(),  // 加购UV
                accumulator.favorMids.size()  // 收藏UV
        );

//        System.out.println("最终统计结果: " + result.toString());
        out.collect(result);
    }
}